package redis.gambing;

import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisLookupMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;

import static redis.gambing.GambleFinalString.*;

/**
 * 该类用来从 Redis 的 HASH_GAMBLER_2_TIMES 中根据 GAMBLER_ID 获取赌徒的参赌次数
 */
public class GamblerTimesInfoLookupMapper implements RedisLookupMapper {
    private final RedisDataTypeDescription description;

    public GamblerTimesInfoLookupMapper() {
        description = new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.HASH, HASH_GAMBLER_2_TIMES);
    }

    /**
     * 获取ITuple中的数据，并将其与新获取的HASH_GAMBLER_2_TIMES一起发送到下游
     */
    @Override
    public List<Values> toTuple(ITuple iTuple, Object o) {
        String gamblerId = getKeyFromTuple(iTuple);
        Long money = iTuple.getLongByField(HASH_GAMBLER_1_MONEY);
        List<Values> values = new ArrayList<>();
        values.add(new Values(gamblerId, money, Short.parseShort(String.valueOf(o))));
        return values;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields(GAMBLER_ID, HASH_GAMBLER_1_MONEY, HASH_GAMBLER_2_TIMES));
    }

    @Override
    public RedisDataTypeDescription getDataTypeDescription() {
        return description;
    }

    @Override
    public String getKeyFromTuple(ITuple iTuple) {
        return iTuple.getStringByField(GAMBLER_ID);
    }

    @Override
    public String getValueFromTuple(ITuple iTuple) {
        return null;
    }
}
